﻿
using System;
using System.IO;
using System.Xml;
using System.Text;
using System.Net;
using System.Collections;
using System.Threading;
using System.Runtime.InteropServices;
using System.Diagnostics;

using Microsoft.BizTalk.Streaming;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.BizTalk.Component.Interop;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.Samples.BizTalk.Adapter.Common;
using Microsoft.XLANGs.BaseTypes;
using Sixeyed.CacheAdapter.Management;
using Sixeyed.CacheAdapter.Interfaces;
using System.Collections.Generic;
using System.Xml.Serialization;
using System.Runtime.Serialization.Formatters.Binary;
using Sixeyed.CacheAdapter.Configuration;
using Sixeyed.CacheAdapter.Logging;
using Sixeyed.CacheAdapter.Helpers;

namespace Sixeyed.CacheAdapter.Runtime
{
    /// <summary>
    /// There is one instance of CacheTransmitterEndpoint class for each every static send port.
    /// Messages will be forwarded to this class by AsyncTransmitterBatch (via CacheTransmitAdapterBatch)
    /// </summary>
    /// <remarks>
    /// For two-way send ports, this is a GET from cache, using the cache key specified in context, or
    /// generating the key from the incoming message. For one-way send ports, this is a PUT to cache, 
    /// storing the incoming message using the cache key from context.
    /// </remarks>
    internal class CacheTransmitterEndpoint: AsyncTransmitterEndpoint
    {
        private IBaseMessage solicitMsg = null;
        private IBTTransportProxy transportProxy = null;
        private AsyncTransmitter asyncTransmitter = null;
        private string propertyNamespace;

        private static object _syncLock = new object();
        private static Dictionary<string, ICacheProvider> _cacheProviders;
        private static Dictionary<string, ICacheProvider> CacheProviders
        {
            get
            {
                if (_cacheProviders == null)
                {
                    lock (_syncLock)
                    {
                        if (_cacheProviders == null)
                        {
                            _cacheProviders = new Dictionary<string, ICacheProvider>();
                        }
                    }
                }
                return _cacheProviders;
            }
        }

        /// <summary>
        /// Constructor with known state
        /// </summary>
        /// <param name="asyncTransmitter">Transmitter</param>
        public CacheTransmitterEndpoint(AsyncTransmitter asyncTransmitter) : base(asyncTransmitter)
        {
            this.asyncTransmitter = asyncTransmitter;
            this.transportProxy = asyncTransmitter.TransportProxy;
        }

        /// <summary>
        /// Opens the endpoint instance
        /// </summary>
        /// <param name="endpointParameters">Parameters</param>
        /// <param name="handlerPropertyBag">Properties</param>
        /// <param name="propertyNamespace">Namespace</param>
        public override void Open(EndpointParameters endpointParameters, IPropertyBag handlerPropertyBag, string propertyNamespace)
        {
            this.propertyNamespace = propertyNamespace;
        }

        /// <summary>
        /// Process incoming message - Put to cache if one-way, retrieve from cache if two-way
        /// </summary>
        public override IBaseMessage ProcessMessage(IBaseMessage message)
        {
            this.solicitMsg = message;
            IBaseMessage responseMsg = null;
            CacheTransmitProperties properties = new CacheTransmitProperties(message, this.propertyNamespace);
            MessageCacheProperties contextProperties = MessageCacheProperties.LoadFromMessage(message);            

            if (!properties.IsTwoWay)
            {
                ICacheProvider provider = GetCacheProvider(properties);

                //one-way send is a PUT to cache, if key is available:
                if (string.IsNullOrEmpty(contextProperties.MessageCacheKey))
                {
                    contextProperties = MessageCacheProperties.LoadFromCache(provider, message);
                }
                if (!string.IsNullOrEmpty(contextProperties.MessageCacheKey))
                {
                    PutToCacheProvider(this.solicitMsg, provider, contextProperties);
                }
                //if message properties were cached, they're no longer needed, remove:
                if (!string.IsNullOrEmpty(contextProperties.InterchangeId))
                {
                    provider.Remove(contextProperties.InterchangeId);
                }
            }
            else
            {
                //solicit-response is a GET from cache if caching is enabled for the message:
                if (contextProperties.IsMessageCacheEnabled)
                {
                    ICacheProvider provider = GetCacheProvider(properties);
                    Stream response = GetFromCacheProvider(this.solicitMsg, provider, contextProperties);
                    responseMsg = BuildResponseMessage(response, this.solicitMsg);            
                    MessageCacheProperties.Save(responseMsg, contextProperties, provider);
                }
            }

            return responseMsg;
        }

        /// <summary>
        /// Puts a byte array representation of the message data to the cache
        /// </summary>
        /// <param name="requestMessage">Message to cache</param>
        /// <param name="provider">Cache provider</param>
        /// <param name="contextProperties">Cache properties</param>
        private void PutToCacheProvider(IBaseMessage requestMessage, ICacheProvider provider, MessageCacheProperties contextProperties)
        {
            IBaseMessagePart requestBodyPart = requestMessage.BodyPart;
            if (requestBodyPart != null)
            {
                Stream requestStream = requestBodyPart.GetOriginalDataStream();
                if (requestStream != null)
                {
                    //store in cache as byte array:                    
                    byte[] request = StreamHelper.GetBytes(requestStream);
                    if (contextProperties.MessageCacheLifespan != TimeSpan.Zero)
                    {
                        provider.Put(contextProperties.MessageCacheKey, request, contextProperties.MessageCacheLifespan);
                    }
                    else
                    {
                        provider.Put(contextProperties.MessageCacheKey, request);
                    }
                }
            }            
        }            

        /// <summary>
        /// Returns message with populated body part
        /// </summary>
        /// <param name="response">Message body data stream</param>
        /// <param name="requestMessage">Message</param>
        /// <returns>Populated message</returns>
        private IBaseMessage BuildResponseMessage(Stream response, IBaseMessage requestMessage)
        {
            requestMessage.BodyPart.Data = response;
            return requestMessage;
        }

        /// <summary>
        /// Returns the configured, initialized cache provider
        /// </summary>
        /// <param name="properties">Cache instance properties</param>
        /// <returns>Cache provider</returns>
        private ICacheProvider GetCacheProvider(CacheTransmitProperties properties)
        {
            if (!CacheProviders.ContainsKey(properties.TransmitLocation.Name))
            {
                ICacheProviderConfiguration config = properties.TransmitLocation.GetCacheConfiguration();
                ICacheProvider provider = CacheFactory.GetProvider(properties.TransmitLocation.Provider, config);
                if (provider != null)
                {
                    CacheProviders.Add(properties.TransmitLocation.Name, provider);
                }
            }
            return CacheProviders[properties.TransmitLocation.Name];
        }

        /// <summary>
        /// Checks the cache for a response message based on the key of the request message.
        /// Returns the cached response message, if available, or the original request.
        /// </summary>
        /// <param name="requestMessage">Incoming request message</param>
        /// <param name="cacheProvider">Cache provider</param>
        /// <param name="contextProperties">Message cache properties</param>
        /// <returns>Cached response, or original request</returns>
        private Stream GetFromCacheProvider(IBaseMessage requestMessage, ICacheProvider cacheProvider, MessageCacheProperties contextProperties)
        {  
            //default response to original request:
            IBaseMessagePart requestBodyPart = requestMessage.BodyPart;
            Stream requestStream = null;
            if (requestBodyPart != null)
            {
                requestStream = requestBodyPart.GetOriginalDataStream();
            }

            //we use the request stream to generate a cache key, 
            //but only if the stream can be reset afterwards:
            Stream responseStream = requestStream;
            if (responseStream.CanRead && responseStream.CanSeek)
            {
                //if the cache key is not provided, get key 
                //from the incoming message body content:
                if (string.IsNullOrEmpty(contextProperties.MessageCacheKey) && requestStream != null)
                {
                    byte[] objectBuffer = StreamHelper.GetBytes(requestStream);
                    contextProperties.MessageCacheKey = cacheProvider.GetCacheKey(objectBuffer);
                }

                if (!string.IsNullOrEmpty(contextProperties.MessageCacheKey))
                {
                    //check for the object in cache:
                    byte[] response = cacheProvider.Get<byte[]>(contextProperties.MessageCacheKey);
                    if (response != null)
                    {
                        contextProperties.IsCached = true;
                        responseStream = new MemoryStream(response);
                    }
                }
                responseStream.Position = 0;
            }
            else 
            {
                Logger.Log(LogLevel.Warn, "CacheTransmitterEndpoint - attempt to cache unseekable stream. Cache disabled for message.");
            }

            return responseStream;
        }
    }
}
